package cep;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternSelectFunction;
import org.apache.flink.cep.PatternStream;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.util.List;
import java.util.Map;
import java.util.Objects;

/**
 * cep步骤逻辑之间 有 严格模式
 * 单步过滤条件 where or until subtype
 * SimpleCondition过滤一条   iterativeConditions通过上一步
 */
public class TestCep {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());
        env.setParallelism(1);
        DataStreamSource<String> dataStreamSource = env.readTextFile("data/cep01.txt");

        SingleOutputStreamOperator<UserRegist> ds = dataStreamSource.map(new MapFunction<String, UserRegist>() {
            @Override
            public UserRegist map(String value) throws Exception {
                String[] arr = value.split(" ");
                return new UserRegist(Integer.valueOf(arr[0]), arr[1], arr[2]);
            }
        });

        //先分组 再各个组 各自匹配
       // KeyedStream<UserRegist, String> keyds = ds.keyBy(f -> f.name);

        //匹配方法
        Pattern<UserRegist, UserRegist> pattern = Pattern.<UserRegist>begin("match_regist").where(new SimpleCondition<UserRegist>() {
            @Override
            public boolean filter(UserRegist value) throws Exception {
                return value.action.equals("regist");
            }
        }).times(5);
        //次数


        //匹配
        PatternStream<UserRegist> ps = CEP.pattern(ds, pattern);

        //select
        //flatSelect
        //process

        ps.select(new PatternSelectFunction<UserRegist,String>() {
            @Override//          map那步匹配到的
            public String select(Map<String, List<UserRegist>> map) throws Exception {
                List<UserRegist> match_regist = map.get("match_regist");
                if(match_regist!=null&&match_regist.size()>0){
                   return  match_regist.get(match_regist.size()-1).name;
                }else{
                    return " ";
                }

            }
        }).print();



        env.execute();
    }

    public static class UserRegist{
        private int id;
        private String name;
        private String action;

        public UserRegist(int id, String name, String action) {
            this.id = id;
            this.name = name;
            this.action = action;
        }

        public int getId() {
            return id;
        }

        public String getName() {
            return name;
        }

        public String getAction() {
            return action;
        }

        public void setId(int id) {
            this.id = id;
        }

        public void setName(String name) {
            this.name = name;
        }

        public void setAction(String action) {
            this.action = action;
        }

        @Override
        public boolean equals(Object o) {
            if (this == o) return true;
            if (!(o instanceof UserRegist)) return false;
            UserRegist that = (UserRegist) o;
            return getId() == that.getId() && getName().equals(that.getName()) && getAction().equals(that.getAction());
        }

        @Override
        public int hashCode() {
            return Objects.hash(getId(), getName(), getAction());
        }
    }
}
